package com.shujia.tour

import com.shujia.util.SparkTool
import com.shujia.util.Config
import org.apache.spark.sql.{SQLContext, SaveMode}
import com.shujia.util.SSXRelation
import com.shujia.util.Geography
import com.shujia.bean.CaseClass.Province

object MakeProvinceApp extends SparkTool {
  /**
    * 编写spark业务逻辑
    *
    */
  override def run(args: Array[String]): Unit = {
    val staypointPath = s"${Config.getString("staypoint.path")}day_id=$day_id"
    val usertagPath = s"${Config.getString("usertag.path")}month_id=201805"
    val provincePath = s"${Config.getString("province.path")}day_id=$day_id"

    LOGGER.info("停留表输入路径：" + staypointPath)
    LOGGER.info("用户画像表输入路径：" + usertagPath)
    LOGGER.info("省游客表输出路径：" + provincePath)

    val sQLContext = new SQLContext(sc)

    import sQLContext.implicits._

    val staypointDF = sQLContext.read.parquet(staypointPath)
    val userTagDF = sQLContext.read.parquet(usertagPath)


    /**
      * 省有课过滤
      * 1、省内停留时间大于3小时
      * 2、最远出行距离大于10km
      *
      * select * from a join b a a.id=b.id where a.gender=‘男’
      * sql会先执行where   下沉过滤器
      */

    //将区县获取省编号的mao集合广播
    val countToProBro = sc.broadcast(SSXRelation.COUNTY_PROVINCE)

    //1、省内停留时间大于3小时
    val stayPointFilterRDD = staypointDF
      .rdd
      .map(row => {
        val mdn = row.getAs[String]("mdn")
        val county_id = row.getAs[String]("county_id")
        //网格编号
        val grid_id = row.getAs[String]("grid_id")
        //通过区县编号获取省编号
        val provinceId = countToProBro.value.get(county_id)

        //每个网格的停留时间
        val duration = row.getAs[String]("duration").toInt

        //手机号和省编码作为key  停留时间和网格编号作为vakue
        (s"${mdn}_$provinceId", (duration, grid_id))
      })
      //统计省内停留时间
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + "-" + y._2))

      //获取停留时间大于3小时的数据
      .filter(_._2._1 > 3 * 60)
      .map(kv => {
        val split = kv._1.split("_")
        val mdn = split(0)
        val provinceId = split(1)
        //省内停留时间
        val sumduration = kv._2._1
        val gridList = kv._2._2

        (mdn, provinceId + "_" + sumduration + "_" + gridList)
      })

    val userTagRDD = userTagDF
      .rdd
      .map(row => {
        val mdn = row.getAs[String]("mdn")
        //常住地网格
        val resi_grid_id = row.getAs[String]("resi_grid_id")
        //常住地区县
        val resi_county_id = row.getAs[String]("resi_county_id")


        (mdn, resi_grid_id + "_" + resi_county_id)
      })


    val resultDF = stayPointFilterRDD
      .join(userTagRDD)
      .map(kv => {
        val mdn = kv._1
        val resi = kv._2._2.split("_")
        val resi_grid_id = resi(0)
        val resi_county_id = resi(1)

        val split = kv._2._1.split("_")
        val provinceId = split(0)
        val sumduration = split(1)
        val gridList = split(2)


        //计算最大出游距离
        val d_max_distance = gridList
          .split("-")
          .toList
          .map(grid => {
            //计算两个网格之间的距离
            Geography.calculateLength(grid.toLong, resi_grid_id.toLong)
          })
          .max //获取最大距离

        s"$mdn\t$resi_county_id\t$provinceId\t$sumduration\t$d_max_distance"

        Province(mdn, resi_county_id, provinceId, sumduration, d_max_distance.toString)
      })
      //2、最远出行距离大于10km
      .filter(_.d_max_distance.toDouble > 10 * 1000)
      .toDF()

    //保存数据
    resultDF
      .write
      .mode(SaveMode.Overwrite)
      .parquet(provincePath)


  }

  /**
    * 初始化方法，在子类设置spark运行时需要的参数
    * conf.setMaster("local")
    */
  override def init(): Unit = {

  }
}
